package common.dao.elastic;

import com.fasterxml.jackson.databind.JsonNode;
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.google.gson.JsonObject;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.search.sort.SortBuilder;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.index.query.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Component;
import common.config.IndexTypeNames;

import javax.inject.Inject;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class ElasticSearchDao {
    final Logger logger = LoggerFactory.getLogger(ElasticSearchDao.class);

    String[] includes = {"*"};
    String[] excludes = {"Indexdata"} ; //, "@id", "@timestamp"};

    // 绝对路径
    final public String tcpPath = "/elastic_admin/tcp_servers.json";

    final private String indexName = IndexTypeNames.MediaDB.getIndexName();//"media";
    final private String[] typeNames = IndexTypeNames.MediaDB.getTypeNames();//{"program" , "sequence" , "scene", "shot"};


    private ElasticAdministratorTransHelper transClient;
    private TransportClient esClient;
    private BulkProcessor bulkProcessor;
    //private ObjectMapper mapper;

    /*************************************
     * use inject for product server
     */
    //@Autowired
    //@Inject
    public ElasticSearchDao() {
        startDao();
    }



    /*************************************
     * don't use inject , used for local backup
     * @param mode
     */
    public ElasticSearchDao(String mode){
        if(mode.equals("debug")){
            startDao();
        }
        else{
            //this.ElasticSearchDao();
            startDao();
        }
    }



    public ElasticSearchDao(String cluster_name, String tcp_host, Integer port){
        this.transClient = new ElasticAdministratorTransHelper();
        this.esClient = this.transClient.getTransportClient(cluster_name, tcp_host, port);

        if(this.esClient != null) logger.info("Connected with Elasticsearch");
        if(this.esClient == null) logger.warn("can not establish Connnection with Elasticsearch");
        this.bulkProcessor = BulkProcessor.builder( this.esClient,new MyBulkProcessorListener() ).build();
        logger.info("[startDao] successfully!");


    }




    @Override
    public void finalize() throws Throwable {
        super.finalize();
        stopDao();
    }

    /*********************************
     * MyBulkProcessorListener for BulkProcessor
     */
    public static class  MyBulkProcessorListener implements BulkProcessor.Listener{
        final Logger logger = LoggerFactory.getLogger(MyBulkProcessorListener.class);
        @Override
        public void beforeBulk(long executionId, BulkRequest request) {
            logger.debug("going to execute bulk of {} requests", request.numberOfActions());
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            logger.debug("bulk executed {} failures", response.hasFailures() ? "with" : "without");
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            logger.warn("error while executing bulk", failure);
        }
    }

    /**************************************
     *
     */
    private void startDao(){
        //JsonFactory factory = new JsonFactory();
        //this.mapper = new ObjectMapper(factory);
        this.transClient = new ElasticAdministratorTransHelper(tcpPath);
        this.esClient = this.transClient.getTransportClient();
        if(this.esClient != null) logger.info("Connected with Elasticsearch");
        if(this.esClient == null) logger.warn("can not establish Connnection with Elasticsearch");
        this.bulkProcessor = BulkProcessor.builder( this.esClient,new MyBulkProcessorListener() ).build();
        logger.info("[startDao] successfully!");
    }

    private void stopDao(){
        this.esClient.close();
    }


    /******************************************
     *
     * @return
     */
    public TransportClient getESClient() {return esClient;}



    /*
    ******************************************************
    * https://www.elastic.co/guide/en/elasticsearch/client/java-api/1.6/index.html
     */

    public boolean save_bulk(String indexName, String typeName, JsonNode doc) {
        boolean ret = false;
        try{
            String id = doc.get("_id").asText();
            String json = doc.toString();
            byte[] bytes = json.getBytes();
            logger.info("[ElasticSearchDao] prepare save_bulk id = " + id);
            //byte[] bytes = mapper.writeValueAsBytes(doc);
            bulkProcessor.add(new IndexRequest(indexName, typeName, id).source(bytes));
            bulkProcessor.flush();
            logger.info("[ElasticSearchDao] success save_bulk id = " + id);
            ret = true;
        } catch (Exception e){
            logger.error("[ElasticSearchDao] failed to save_bulk doc");
        }
        return ret;

    }



    public void delete_bulk(String indexName, String typeName, String id) {
        bulkProcessor.add(new DeleteRequest(indexName, typeName, id));
        logger.info("[ElasticSearchDao] delete id = " + id);
    }


    public boolean delete(String indexName, String typeName, String id){
        boolean ret = false;
        try{
            DeleteResponse response = this.esClient.prepareDelete(indexName, typeName, id)
                    .execute()
                    .actionGet();
            ret = true;
        }
        catch (Exception e){
            logger.error("[delete error] msg = {}", e.getMessage());
        }
        return ret;
    }



    public boolean index(String indexName, String typeName, String id,JsonNode doc){
        boolean ret = false;
        try{
            //String id = doc.get("_id").asText();
            logger.info("[index] id = {}", id);
            String json = doc.toString();
            IndexResponse response = this.esClient.prepareIndex(indexName, typeName, id)
                    .setSource(json)
                    .execute()
                    .actionGet();
            // Index name
            String _index = response.getIndex();
            // Type name
            String _type = response.getType();
            // Document ID (generated or not)
            String _id = response.getId();
            // Version (if it's the first time you index this document, you will get: 1)
            long _version = response.getVersion();
            // isCreated() is true if the document is a new one, false if it has been updated
            boolean created = response.isCreated();
            logger.info("[index] _index = {}, _type = {}, _id = {}, _version = {}", _index,_type,_id,_version);
            ret = true;
        }
        catch (Exception e){
            logger.error("[ElasticSearchDao] failed to index doc");
            logger.error("[ERROR_MSG = ] {}", e.getMessage());
        }
        return ret;
    }


    public IndexResponse indexAndResponse(String indexName, String typeName, String id,JsonNode doc){
        boolean ret = false;
        IndexResponse response = null;
        try{
            //String id = doc.get("_id").asText();
            logger.info("[index] id = {}", id);
            String json = doc.toString();
            response = this.esClient.prepareIndex(indexName, typeName, id)
                    .setSource(json)
                    .execute()
                    .actionGet();
            // Index name
            String _index = response.getIndex();
            // Type name
            String _type = response.getType();
            // Document ID (generated or not)
            String _id = response.getId();
            // Version (if it's the first time you index this document, you will get: 1)
            long _version = response.getVersion();
            // isCreated() is true if the document is a new one, false if it has been updated
            boolean created = response.isCreated();
            logger.info("[index] _index = {}, _type = {}, _id = {}, _version = {}", _index,_type,_id,_version);
            ret = true;
        }
        catch (Exception e){
            logger.error("[ElasticSearchDao] failed to index doc");
            logger.error("[ERROR_MSG = ] {}", e.getMessage());
        }
        return response;
    }



    /*
    https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html
     */
    public void update(){

    }

    /***********************************************
     *
     * @param level
     * @param id
     * @return
     */
    public GetResponse get(String level, String id){
        GetRequestBuilder request = esClient.prepareGet(indexName, level, id);
        request.setFetchSource(includes, excludes);
        logger.debug("request = /{}/{}/{}", indexName, level, id);

        GetResponse response =
                request.execute().actionGet();

        logger.debug("elasticsearch get by id: {} \n {} " , id, response.isExists());

        //String result_str = response.getSourceAsString();
        //JsonNode result_node = JsonToJavaObject.loadJsonNode(result_str);
        //((ObjectNode)result_node).remove("Indexdata");
        return response;
    }


    public GetResponse getApi(String index,String type, String id){
        GetRequestBuilder request = esClient.prepareGet(index, type, id);
        //request.setFetchSource(includes, excludes);
        logger.debug("request = /{}/{}/{}", index, type, id);

        GetResponse response =
                request.execute().actionGet();

        logger.debug("elasticsearch get by id: {} \n {} " , id, response.isExists());

        //String result_str = response.getSourceAsString();
        //JsonNode result_node = JsonToJavaObject.loadJsonNode(result_str);
        //((ObjectNode)result_node).remove("Indexdata");
        return response;
    }

    /*******************************************************
     *
     * @param query
     * @param from
     * @param size
     * @param sort
     * @param levels
     * @param is_explain
     * @return
     */
    public SearchRequestBuilder genSearchRequset(QueryBuilder query, Integer from, Integer size, SortBuilder sort,List<String> levels, Boolean is_explain){

        SearchRequestBuilder request = esClient.prepareSearch(indexName);

        int levelCount = levels.size();
        logger.debug("levels size : {} \n" , levelCount);
        if(levelCount == 1) request.setTypes(levels.get(0));
        else if(levelCount == 2) request.setTypes(levels.get(0),levels.get(1));
        else if(levelCount == 3) request.setTypes(levels.get(0),levels.get(1),levels.get(2));
        else if(levelCount == 4) request.setTypes(levels.get(0),levels.get(1),levels.get(2),levels.get(3));


        request.setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                .setQuery(query)
                //.setPostFilter(filter)
                .setFrom(from)
                .setSize(size)
                .addSort(sort)
                .setExplain(is_explain)
                .setFetchSource(includes,excludes);

        logger.debug("elasticsearch request: \n {}", request.toString());
        return request;
    }


    /**
     * 探索  request  支持的API
     * @param query
     * @return
     */
    public SearchRequestBuilder genPossibleSearchRequest(QueryBuilder query){
        SearchRequestBuilder request = esClient.prepareSearch(indexName);
        request.setSearchType("type");
        request.setQuery(query);

        Map<String, String> aggs = new HashMap<>();
        request.setAggregations(aggs);

        request.setHighlighterType("type");

        request.setMinScore(0.001f);


        return request;

    }


    public SearchResponse searchWithDsl(String index,String type, String dsl,Integer from, Integer size){
        WrapperQueryBuilder query = QueryBuilders.wrapperQuery( dsl );
        SearchRequestBuilder request = this.esClient.prepareSearch(index)
                .setTypes(type)
                .setQuery( query )
                //.addFacet(FacetBuilders.statisticalFacet("age").field("timestamp"))
                .setFrom(from)
                .setSize(size);
        SearchResponse response =
                request.execute().actionGet();
        return response;
    }

    public SearchRequestBuilder genSearchWithApiRequest(String index,String type,QueryBuilder query,Integer from, Integer size,SortBuilder sort, Boolean is_explain){
        String[] includes = {"*"}; //{"text", "entities", "contexts"};
        String[] excludes = {};
        SearchRequestBuilder request = this.esClient.prepareSearch(index)
                .setTypes(type)
                .setQuery( query )
                .setFrom(from)
                .setSize(size)
                .setFetchSource(includes, excludes);
        if(sort != null)
            request.addSort( sort );
        if(is_explain != null)
            request.setExplain( is_explain );

        logger.info("\n\n[genSearchWithApiRequest] request = \n{}\n\n", request.toString());

        return request;
    }

    public SearchResponse searchWithApi( String index,String type,QueryBuilder query,Integer from, Integer size,SortBuilder sort, Boolean is_explain){

        SearchRequestBuilder request = genSearchWithApiRequest(index,type,query,from,size,sort,is_explain);

        logger.info("\n\n[searchWithApi] request = \n{}\n\n", request.toString());

        SearchResponse response =  request.execute().actionGet();

        return response;
    }


    /************************************************
     *
     * @param query
     * @param from
     * @param size
     * @param sort
     * @param levels
     * @return
     */
    public SearchResponse search(QueryBuilder query, Integer from, Integer size, SortBuilder sort,List<String> levels, Boolean is_explain) {

        SearchRequestBuilder request = genSearchRequset(query,from,size,sort,levels,is_explain);

        SearchResponse response =
                request.execute().actionGet();

        logger.debug("elasticsearch response: {} hits", response.getHits().totalHits());
        logger.trace("elasticsearch response: {} hits", response.toString());
        return response;
    }

    /********************************************
     *
     * @param template
     * @param params
     * @return
     */
    public SearchResponse searchWithTemplate(String template,Map<String, Object> params){
        logger.debug("elasticsearch query template :\n {}" , template);
        logger.debug("elasticsearch query params   :\n {}" , params.toString() );

        esClient.preparePutIndexedScript("mustache", "random_template",
                template
        ).get();

        SearchResponse searchResponse = esClient.prepareSearch("media")
                .setTypes("video")
                .setTemplateName("random_template")
                .setTemplateType(ScriptService.ScriptType.INDEXED)
                .setTemplateParams(params)
                .get();

        logger.debug("elasticsearch response: {} hits", searchResponse.getHits().totalHits());
        logger.trace("elasticsearch response: {} hits", searchResponse.toString() );

        return searchResponse;
    }

    /*********************************************
     *
     * @param text
     * @param analyzer
     * @return
     */
    public AnalyzeResponse analyze(String text, String analyzer){
        logger.info("[analyze] analyzer = {}, text = {}", analyzer, text);
        this.esClient.admin().indices().prepareAnalyze("trying out").execute().actionGet();
        IndicesAdminClient adminClient = this.esClient.admin().indices();
        String index = IndexTypeNames.MediaDB.getIndexName();
        AnalyzeRequestBuilder request = adminClient.prepareAnalyze(text).setIndex(index).setAnalyzer(analyzer);
        AnalyzeResponse response = request.execute().actionGet();

        //AnalyzeRequest analyzeRequest = new AnalyzeRequest(text);

        logger.debug("elasticsearch response: {} hits", response.getTokens().size());
        //logger.trace("elasticsearch response: {} hits", response.getTokens().size());
        return response;
    }






    /***********************************************************************************************
     *
     * @param args
     */
    public static void main(String[] args){

        test_search_with_dsl();

    }








    public static void test_save(){
        ElasticSearchDao dao = new ElasticSearchDao("local_cluster", "127.0.0.1", 9300); //("debug");
        String json = "{" +
                "\"user\":\"kimchy\"," +
                "\"postDate\":\"2013-01-30\"," +
                "\"message\":\"trying out Elasticsearch\"" +
                "}";

        IndexResponse response = dao.esClient.prepareIndex("twitter", "tweet")
                .setSource(json)
                .execute()
                .actionGet();
    }

    public static void test_analyser(){
        /*
        * backup analyzer
         */
        ElasticSearchDao dao = new ElasticSearchDao("debug");
        AnalyzeResponse analyzeResponse = dao.analyze("北京市车流","ik_smart");
        for (AnalyzeResponse.AnalyzeToken analyzeToken : analyzeResponse.getTokens()) {
            //System.out.println(analyzeToken.);
            System.out.println(
                    "term: " + analyzeToken.getTerm() + "\n"
                            + "type: " + analyzeToken.getType() + "\n"
                            + "position: " + analyzeToken.getPosition() + "\n"
                            + "start: " + analyzeToken.getStartOffset() + "\n"
                            + "end: " + analyzeToken.getEndOffset() + "\n"
            );
        }
    }



    public static void test_search_with_dsl(){
        /*
       {
      "query": {
        "bool" : {
          "must" : {
             "match":{"text":"Eugene"}
           },
           "must" : {
             "match":{"entities":"YAGO:NASA"}
           },
           "must" : {
             "match" : {"entities" : "YAGO:World_War_II"}
           }
        }
      }
    }
         */
        ElasticSearchDao elasticSearchDao = new ElasticSearchDao();
        String dsl = String.format(
            "{\n" +
                    "  \"query\":{\n" +
                    "    \"match\":{\n" +
                    "       \"category\" : \"California\"\n" +
                    "    }\n" +
                    "  }\n" +
                    "}"
        );
        //WrapperQueryBuilder query = QueryBuilders.wrapperQuery(dsl);
        SearchResponse response = elasticSearchDao.searchWithDsl("enwikinews-doc", "doc", dsl, 0, 10);
        System.out.println(response.toString());

    }

}
